package com.such.kit.database.bean;

import java.io.ByteArrayInputStream;
import java.io.FileOutputStream;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import javax.sql.DataSource;
import org.apache.tools.ant.filters.StringInputStream;
import com.alibaba.fastjson.JSON;
import com.such.kit.database.JdbcCommons;
import com.such.kit.database.MetaDataCommons;
import com.such.kit.database.dialect.DDLFactory;
import com.such.kit.database.dialect.DialectCommons;
import com.such.kit.database.exception.DatabaseException;
import com.such.kit.datacast.CastCommons;
import com.such.kit.easyparam.ParamList;
import com.such.kit.easyparam.ParamMap;
import com.such.kit.file.FileCommons;
import com.such.kit.file.JarFileCommons;
import com.such.kit.io.bean.SqlIOConfig;
import com.such.kit.io.template.JSONIOTemplate;
import com.such.kit.io.template.SqlIOTemplate;
import com.such.kit.io.template.XmlIOTemplate;
import com.such.kit.io.source.DatabaseIOSource;
import com.such.kit.io.source.MemoryIOSource;
import com.such.kit.placeholder.PlaceholderCommons;
import com.such.kit.stream.StreamCommons;
import com.such.kit.string.StringCommons;
import com.such.kit.validate.ValidateCommons;

public class MigrationExtractResult {

	public static final String ENLARGE_NAME_XML = "xml";
	public static final String ENLARGE_NAME_JSON = "json";
	public static final String ENLARGE_NAME_SQL = "sql";
	public static final String ENLARGE_NAME_BIG_DATA = "bd";
	public static final String FILE_SOURCE_EXPRESSION = "migration_#[datetime]";
	public static final String DEFINITION_FILE_TEMPLATE = "com/such/kit/database/migration/configs/definitionTemplate.xml";
	public static final String DEFINITION_FILE_NAME = "definition";
	public static final String BIG_DATA_INDEX_FILE_TEMPLATE = "com/such/kit/database/migration/configs/bigDataIndexTemplate.json";
	public static final String BIG_DATA_INDEX_KEY_UPDATE_STATEMENT = "updateStatement";
	public static final String BIG_DATA_INDEX_KEY_FILE_NAME = "fileName";
	public static final String BIG_DATA_INDEX_KEY_LOB_TYPE = "lobType";
	public static final String DATA_FOLDER_NAME = "data";

	public static final String IO_EXPRESSION_DATABASE = "database";
	public static final String IO_EXPRESSION_TABLE = "table";
	public static final String IO_EXPRESSION_COLUMN = "#[schema]_#[tableName]_columns";
	public static final String IO_EXPRESSION_PRIMARY_KEY = "#[schema]_#[tableName]_primaryKeys";
	public static final String IO_EXPRESSION_PRIMARY_KEY_PK_COLUMN_NAME = "#[schema]_#[tableName]_primaryKeys_#[pkName]_pkColumnNames";
	public static final String IO_EXPRESSION_FOREIGNKEY = "#[schema]_#[tableName]_foreignKeys";
	public static final String IO_EXPRESSION_FOREIGNKEY_FK_COLUMN_NAME = "#[fkSchema]_#[fkTableName]_foreignKeys_#[fkName]_fkColumnNames";
	public static final String IO_EXPRESSION_FOREIGNKEY_REFERENCE_COLUMN_NAME = "#[fkSchema]_#[fkTableName]_foreignKeys_#[fkName]_referenceColumnNames";
	public static final String IO_EXPRESSION_INDEX = "#[schema]_#[tableName]_indexs";
	public static final String IO_EXPRESSION_INDEX_COLUMN_NAME = "#[schema]_#[tableName]_indexs_#[indexName]_columnNames";
	public static final String IO_EXPRESSION_BIG_DATA_INDEX = "bigDataIndex";

	private static final int BIG_DATA_PAGE_LIMIT = 100;

	private MigrationConfig config;
	private String workDir;
	private String fileSource;
	private String fileSourcePath;
	private String definitionFilePath;
	private String definitionScriptPath;
	private String dataFolderPath;
	private List<DDLTableKey> tableKeys = new ArrayList<DDLTableKey>();
	private Map<DDLTableKey, DDLTable> tablesMap = new HashMap<DDLTableKey, DDLTable>();

	public MigrationExtractResult(String workDir, MigrationConfig config) throws Exception {
		this(workDir, 
				PlaceholderCommons.replacePlaceholder(FILE_SOURCE_EXPRESSION, new ParamMap<String, String>().putParam("datetime", CastCommons.dateToString(new Date(), CastCommons.DATE_TIME_NUMBER))), 
				config);
	}

	public MigrationExtractResult(String workDir, String fileSource, MigrationConfig config) throws Exception {
		this.config = config;
		this.workDir = workDir;
		this.fileSource = fileSource;
		this.fileSourcePath = FileCommons.appendPathNode(workDir, fileSource);
		this.definitionFilePath = FileCommons.appendPathNode(workDir, fileSource, StringCommons.merger(DEFINITION_FILE_NAME, ".", ENLARGE_NAME_XML));
		this.definitionScriptPath = FileCommons.appendPathNode(workDir, fileSource, StringCommons.merger(DEFINITION_FILE_NAME, ".", ENLARGE_NAME_SQL));
		this.dataFolderPath = FileCommons.appendPathNode(workDir, fileSource, DATA_FOLDER_NAME);
	}

	public MigrationConfig getConfig() {
		return this.config;
	}

	public void setConfig(MigrationConfig config) {
		this.config = config;
	}

	public String getWorkDir() {
		return this.workDir;
	}

	public String getFileSource() {
		return this.fileSource;
	}

	public String getFileSourcePath() {
		return this.fileSourcePath;
	}

	public String getDefinitionFilePath() {
		return this.definitionFilePath;
	}

	public String getDefinitionScriptPath() {
		return this.definitionScriptPath;
	}

	public String getDataFolderPath() {
		return this.dataFolderPath;
	}

	public List<DDLTableKey> getTableKeys() {
		return this.tableKeys;
	}

	public Map<DDLTableKey, DDLTable> getTablesMap() {
		return this.tablesMap;
	}

	public void addTable(DDLTable table) {
		DDLTableKey tableKey = new DDLTableKey(table.getSchema(), table.getTableName());
		this.tableKeys.add(tableKey);
		this.tablesMap.put(tableKey, table);
	}

	public void addTables(List<DDLTable> tables) {
		if (ValidateCommons.isNotEmpty(tables)) {
			for (DDLTable table : tables) {
				addTable(table);
			}
		}
	}

	public void putColumn(DDLTableKey tableKey, DDLColumn column) {
		DDLTable table = this.tablesMap.get(tableKey);
		table.getColumns().add(column);
	}

	public void putColumns(DDLTableKey tableKey, List<DDLColumn> columns) {
		if (ValidateCommons.isNotEmpty(columns)) {
			DDLTable table = this.tablesMap.get(tableKey);
			table.getColumns().addAll(columns);
		}
	}

	public void putPrimaryKey(DDLTableKey tableKey, DDLPrimaryKey primaryKey) {
		DDLTable table = this.tablesMap.get(tableKey);
		table.getPrimaryKeys().add(primaryKey);
	}

	public void putPrimaryKeys(DDLTableKey tableKey, List<DDLPrimaryKey> primaryKeys) {
		if (ValidateCommons.isNotEmpty(primaryKeys)) {
			DDLTable table = this.tablesMap.get(tableKey);
			table.getPrimaryKeys().addAll(primaryKeys);
		}
	}

	public void putForeignKey(DDLTableKey tableKey, DDLForeignKey foreignKey) {
		DDLTable table = this.tablesMap.get(tableKey);
		table.getForeignKeys().add(foreignKey);
	}

	public void putForeignKeys(DDLTableKey tableKey, List<DDLForeignKey> foreignKeys) {
		if (ValidateCommons.isNotEmpty(foreignKeys)) {
			DDLTable table = this.tablesMap.get(tableKey);
			table.getForeignKeys().addAll(foreignKeys);
		}
	}

	public void putIndex(DDLTableKey tableKey, DDLIndex index) {
		DDLTable table = this.tablesMap.get(tableKey);
		table.getIndexs().add(index);
	}

	public void putIndexs(DDLTableKey tableKey, List<DDLIndex> indexs) {
		if (ValidateCommons.isNotEmpty(indexs)) {
			DDLTable table = this.tablesMap.get(tableKey);
			table.getIndexs().addAll(indexs);
		}
	}

	/**
	 * <p>方法名：readIn</p>
	 * <p>描述：从 fileSource 读取文件构建 MigrationExtractResult</p>
	 * @throws Exception 
	 */
	public void readIn() throws Exception {
		if (ValidateCommons.isEmpty(this.fileSource)) {
			throw new DatabaseException("未指定 fileSource");
		}

		if (this.config.getExtractScopes().contains(MigrationConfig.SCOPE_DEFINITION)) {
			// 开始处理定义（definition.xml）
			MemoryIOSource definitionIOSource = new MemoryIOSource();
			XmlIOTemplate definitionIO = new XmlIOTemplate(JarFileCommons.getResourceStream(DEFINITION_FILE_TEMPLATE));
			definitionIO.input(definitionIOSource, this.definitionFilePath);

			/* 读取 database 节点
			List<Map<String, String>> databaseResults = definitionIOSource.getInputResult(IO_EXPRESSION_DATABASE);
			String dbType = databaseResults.get(0).get("dbType");
			*/

			// 读取 table 节点（既所有 table）
			List<Map<String, Object>> tableResults = definitionIOSource.getInputResult(IO_EXPRESSION_TABLE);
			if (ValidateCommons.isNotEmpty(tableResults)) {
				// 遍历所有 table 并构建 DDLTable
				for (Map<String, Object> tableResult : tableResults) {
					DDLTable table = new DDLTable();
					table.setSchema((String) tableResult.get("schema"));
					table.setTableName((String) tableResult.get("tableName"));
					table.setTableType((String) tableResult.get("tableType"));
					table.setTableComment((String) tableResult.get("tableComment"));

					// 处理 table column
					List<Map<String, Object>> columnResults = definitionIOSource.getInputResult(IO_EXPRESSION_COLUMN, tableResult);
					if (ValidateCommons.isNotEmpty(columnResults)) {
						for (Map<String, Object> columnResult : columnResults) {
							DDLColumn column = new DDLColumn();
							column.setSchema((String) columnResult.get("schema"));
							column.setTableName((String) columnResult.get("tableName"));
							column.setColumnName((String) columnResult.get("columnName"));
							column.setDataType((String) columnResult.get("dataType"));
							column.setColumnSize(CastCommons.stringToInteger(StringCommons.emptyValue((String) columnResult.get("columnSize"), "0")));
							column.setDecimalDigits(CastCommons.stringToInteger(StringCommons.emptyValue((String) columnResult.get("decimalDigits"), "0")));
							column.setDefaultValue((String) columnResult.get("defaultValue"));
							column.setNotNull(CastCommons.stringToBoolean((String) columnResult.get("notNull")));
							column.setAutoincrement(CastCommons.stringToBoolean((String) columnResult.get("autoincrement")));
							column.setColumnComment((String) columnResult.get("columnComment"));
							table.addColumn(column);
						}
					}

					// 处理 table primaryKey
					List<Map<String, Object>> primaryKeyResults = definitionIOSource.getInputResult(IO_EXPRESSION_PRIMARY_KEY, tableResult);
					if (ValidateCommons.isNotEmpty(primaryKeyResults)) {
						for (Map<String, Object> primaryKeyResult : primaryKeyResults) {
							DDLPrimaryKey primaryKey = new DDLPrimaryKey();
							primaryKey.setSchema((String) primaryKeyResult.get("schema"));
							primaryKey.setTableName((String) primaryKeyResult.get("tableName"));
							primaryKey.setPkName((String) primaryKeyResult.get("pkName"));

							List<Map<String, Object>> pkColumnNameResults = definitionIOSource.getInputResult(IO_EXPRESSION_PRIMARY_KEY_PK_COLUMN_NAME, primaryKeyResult);
							if (ValidateCommons.isNotEmpty(pkColumnNameResults)) {
								List<String> pkColumnNames = new ArrayList<String>();
								for (Map<String, Object> pkColumnNameResult : pkColumnNameResults) {
									pkColumnNames.add((String) pkColumnNameResult.get("pkColumnName"));
								}
								primaryKey.setPkColumnNames(pkColumnNames);
							}

							table.addPrimaryKey(primaryKey);
						}
					}

					// 处理 table foreignKey
					List<Map<String, Object>> foreignKeyResults = definitionIOSource.getInputResult(IO_EXPRESSION_FOREIGNKEY, tableResult);
					if (ValidateCommons.isNotEmpty(foreignKeyResults)) {
						for (Map<String, Object> foreignKeyResult : foreignKeyResults) {
							DDLForeignKey foreignKey = new DDLForeignKey();
							foreignKey.setFkSchema((String) foreignKeyResult.get("fkSchema"));
							foreignKey.setFkTableName((String) foreignKeyResult.get("fkTableName"));
							foreignKey.setFkName((String) foreignKeyResult.get("fkName"));
							foreignKey.setReferenceSchema((String) foreignKeyResult.get("referenceSchema"));
							foreignKey.setReferenceTableName((String) foreignKeyResult.get("referenceTableName"));
							foreignKey.setReferenceName((String) foreignKeyResult.get("referenceName"));

							List<Map<String, Object>> fkColumnNameResults = definitionIOSource.getInputResult(IO_EXPRESSION_FOREIGNKEY_FK_COLUMN_NAME, foreignKeyResult);
							if (ValidateCommons.isNotEmpty(fkColumnNameResults)) {
								for (Map<String, Object> fkColumnNameResult : fkColumnNameResults) {
									foreignKey.addFkColumnName((String) fkColumnNameResult.get("fkColumnName"));
								}
							}

							List<Map<String, Object>> referenceColumnNameResults = definitionIOSource.getInputResult(IO_EXPRESSION_FOREIGNKEY_REFERENCE_COLUMN_NAME, foreignKeyResult);
							if (ValidateCommons.isNotEmpty(referenceColumnNameResults)) {
								for (Map<String, Object> referenceColumnNameResult : referenceColumnNameResults) {
									foreignKey.addReferenceColumnName((String) referenceColumnNameResult.get("referenceColumnName"));
								}
							}

							table.addForeignKey(foreignKey);
						}
					}

					// 处理 table index
					List<Map<String, Object>> indexResults = definitionIOSource.getInputResult(IO_EXPRESSION_INDEX, tableResult);
					if (ValidateCommons.isNotEmpty(indexResults)) {
						for (Map<String, Object> indexResult : indexResults) {
							DDLIndex index = new DDLIndex();
							index.setSchema((String) indexResult.get("schema"));
							index.setTableName((String) indexResult.get("tableName"));
							index.setIndexName((String) indexResult.get("indexName"));
							index.setUnique(CastCommons.stringToBoolean((String) indexResult.get("unique")));

							List<Map<String, Object>> columnNameResults = definitionIOSource.getInputResult(IO_EXPRESSION_INDEX_COLUMN_NAME, indexResult);
							if (ValidateCommons.isNotEmpty(columnNameResults)) {
								List<String> columnNames = new ArrayList<String>();
								for (Map<String, Object> columnNameResult : columnNameResults) {
									columnNames.add((String) columnNameResult.get("columnName"));
								}
								index.setColumnNames(columnNames);
							}

							table.addIndex(index);
						}
					}

					addTable(table);
				}
			}
		}
	}

	/**
	 * <p>方法名：writeOut</p>
	 * <p>描述：将 MigrationExtractResult 生成文件到 fileSource</p>
	 * @param fromDataSource 来源
	 * @throws Exception 
	 */
	public void writeOut(DataSource fromDataSource) throws Exception {
		if (ValidateCommons.isEmpty(fromDataSource)) {
			throw new DatabaseException("未指定 dataSource");
		}

		if (ValidateCommons.isNotEmpty(this.tableKeys)) {
			FileCommons.createFolder(this.fileSourcePath);
			JdbcCommons jdbcCommons = new JdbcCommons(fromDataSource);
			MetaDataCommons metaDataCommons = jdbcCommons.getMetaDataCommons();
			DialectCommons dialectCommons = jdbcCommons.getDialectCommons();
			String dbType = metaDataCommons.getDbType();

			if (this.config.getExtractScopes().contains(MigrationConfig.SCOPE_DEFINITION)) {
				// 开始处理定义（definition.xml）
				boolean creationDefinitionScript = this.config.isCreationDefinitionScript();
				FileOutputStream definitionScript = null;
				if (creationDefinitionScript) {
					definitionScript = new FileOutputStream(this.definitionScriptPath);
				}
				try {
					MemoryIOSource definitionIOSource = new MemoryIOSource();

					Map<String, Object> databaseResult = new HashMap<String, Object>();
					databaseResult.put("dbType", dbType);
					definitionIOSource.setOutputResult(IO_EXPRESSION_DATABASE, null, databaseResult);

					for (DDLTableKey tableKey : tableKeys) {
						DDLTable table = this.tablesMap.get(tableKey);

						// 处理 schema 映射
						if (this.config.hasTableSchemaRemap()) {
							String schema = table.getSchema();
							table.reset(new ParamMap<String, String>().putParam(schema, this.config.fromRemapTo(schema)), null, null, null);
						}

						if (creationDefinitionScript) {
							String ddl = DDLFactory.toString(new ParamList<String>()
									.addAllParam(metaDataCommons.dropTable(table, false))
									.addAllParam(metaDataCommons.createTable(table, false)));
							StreamCommons.binaryTransport(new StringInputStream(StringCommons.merger(ddl, StringCommons.NEW_LINE, StringCommons.NEW_LINE)), true, definitionScript, false);
						}

						Map<String, Object> tableResult = new HashMap<String, Object>();
						tableResult.put("schema", table.getSchema());
						tableResult.put("tableName", table.getTableName());
						tableResult.put("tableType", table.getTableType());
						tableResult.put("tableComment", table.getTableComment());
						definitionIOSource.setOutputResult(IO_EXPRESSION_TABLE, null, tableResult);

						List<DDLColumn> columns = table.getColumns();
						if (ValidateCommons.isNotEmpty(columns)) {
							for (DDLColumn column : columns) {
								Map<String, Object> columnResult = new HashMap<String, Object>();
								columnResult.put("schema", column.getSchema());
								columnResult.put("tableName", column.getTableName());
								columnResult.put("columnName", column.getColumnName());
								columnResult.put("dataType", column.getDataType());
								columnResult.put("columnSize", CastCommons.numberToString(column.getColumnSize()));
								columnResult.put("decimalDigits", CastCommons.numberToString(column.getDecimalDigits()));
								columnResult.put("defaultValue", column.getDefaultValue());
								columnResult.put("notNull", Boolean.toString(column.getNotNull()));
								columnResult.put("autoincrement", Boolean.toString(column.getAutoincrement()));
								columnResult.put("columnComment", column.getColumnComment());
								definitionIOSource.setOutputResult(IO_EXPRESSION_COLUMN, tableResult, columnResult);
							}
						}

						List<DDLPrimaryKey> primaryKeys = table.getPrimaryKeys();
						if (ValidateCommons.isNotEmpty(primaryKeys)) {
							for (DDLPrimaryKey primaryKey : primaryKeys) {
								Map<String, Object> primaryKeyResult = new HashMap<String, Object>();
								primaryKeyResult.put("schema", primaryKey.getSchema());
								primaryKeyResult.put("tableName", primaryKey.getTableName());
								primaryKeyResult.put("pkName", primaryKey.getPkName());
								definitionIOSource.setOutputResult(IO_EXPRESSION_PRIMARY_KEY, tableResult, primaryKeyResult);

								List<String> pkColumnNames = primaryKey.getPkColumnNames();
								if (ValidateCommons.isNotEmpty(pkColumnNames)) {
									for (String pkColumnName : pkColumnNames) {
										Map<String, Object> pkColumnNameResult = new HashMap<String, Object>();
										pkColumnNameResult.put("pkColumnName", pkColumnName);
										definitionIOSource.setOutputResult(IO_EXPRESSION_PRIMARY_KEY_PK_COLUMN_NAME, primaryKeyResult, pkColumnNameResult);
									}
								}
							}
						}

						List<DDLForeignKey> foreignKeys = table.getForeignKeys();
						if (ValidateCommons.isNotEmpty(foreignKeys)) {
							for (DDLForeignKey foreignKey : foreignKeys) {
								Map<String, Object> foreignKeyResult = new HashMap<String, Object>();
								foreignKeyResult.put("fkSchema", foreignKey.getFkSchema());
								foreignKeyResult.put("fkTableName", foreignKey.getFkTableName());
								foreignKeyResult.put("fkName", foreignKey.getFkName());
								foreignKeyResult.put("referenceSchema", foreignKey.getReferenceSchema());
								foreignKeyResult.put("referenceTableName", foreignKey.getReferenceTableName());
								foreignKeyResult.put("referenceName", foreignKey.getReferenceName());
								definitionIOSource.setOutputResult(IO_EXPRESSION_FOREIGNKEY, tableResult, foreignKeyResult);
								
								List<String> fkColumnNames = foreignKey.getFkColumnNames();
								if (ValidateCommons.isNotEmpty(fkColumnNames)) {
									for (String fkColumnName : fkColumnNames) {
										Map<String, Object> fkColumnNameResult = new HashMap<String, Object>();
										fkColumnNameResult.put("fkColumnName", fkColumnName);
										definitionIOSource.setOutputResult(IO_EXPRESSION_FOREIGNKEY_FK_COLUMN_NAME, foreignKeyResult, fkColumnNameResult);
									}
								}

								List<String> referenceColumnNames = foreignKey.getReferenceColumnNames();
								if (ValidateCommons.isNotEmpty(referenceColumnNames)) {
									for (String referenceColumnName : referenceColumnNames) {
										Map<String, Object> referenceColumnNameResult = new HashMap<String, Object>();
										referenceColumnNameResult.put("referenceColumnName", referenceColumnName);
										definitionIOSource.setOutputResult(IO_EXPRESSION_FOREIGNKEY_REFERENCE_COLUMN_NAME, foreignKeyResult, referenceColumnNameResult);
									}
								}
							}
						}

						List<DDLIndex> indexs = table.getIndexs();
						if (ValidateCommons.isNotEmpty(indexs)) {
							for (DDLIndex index : indexs) {
								Map<String, Object> indexResult = new HashMap<String, Object>();
								indexResult.put("schema", index.getSchema());
								indexResult.put("tableName", index.getTableName());
								indexResult.put("indexName", index.getIndexName());
								indexResult.put("unique", Boolean.toString(index.getUnique()));
								definitionIOSource.setOutputResult(IO_EXPRESSION_INDEX, tableResult, indexResult);

								List<String> columnNames = index.getColumnNames();
								if (ValidateCommons.isNotEmpty(columnNames)) {
									for (String columnName : columnNames) {
										Map<String, Object> columnNameResult = new HashMap<String, Object>();
										columnNameResult.put("columnName", columnName);
										definitionIOSource.setOutputResult(IO_EXPRESSION_INDEX_COLUMN_NAME, indexResult, columnNameResult);
									}
								}
							}
						}
					}

					XmlIOTemplate definitionIO = new XmlIOTemplate(JarFileCommons.getResourceStream(DEFINITION_FILE_TEMPLATE));
					definitionIO.output(definitionIOSource, this.fileSourcePath, DEFINITION_FILE_NAME);
				} finally {
					StreamCommons.close(definitionScript);
				}
			}

			if (this.config.getExtractScopes().contains(MigrationConfig.SCOPE_DATA)) {
				// 开始处理数据
				if (ValidateCommons.isEmpty(fromDataSource)) {
					throw new DatabaseException("未指定 dataSource");
				}
				FileCommons.createFolder(this.dataFolderPath);

				DatabaseIOSource dataIOSource = new DatabaseIOSource(fromDataSource);
				for (DDLTableKey tableKey : this.tableKeys) {
					DDLTable table = this.tablesMap.get(tableKey);
					String schema = table.getSchema();
					String tableName = table.getTableName();
					String fromTableName = tableName;
					String toTableName = tableName;
					if (ValidateCommons.isNotEmpty(schema) && ValidateCommons.isNotEmpty(tableName)) {
						fromTableName = StringCommons.merger(this.config.toRemapFrom(schema), ".", tableName);
						toTableName = StringCommons.merger(this.config.fromRemapTo(schema), ".", tableName);
					}

					StringBuffer primaryKeyWheres = new StringBuffer();
					List<DDLColumn> columns = table.getColumns();
					List<DDLColumn> lobColumns = new ArrayList<DDLColumn>();
					if (ValidateCommons.isNotEmpty(columns)) {
						StringBuffer insertColumnNames = new StringBuffer();
						StringBuffer selectColumnNames = new StringBuffer();
						StringBuffer columnValues = new StringBuffer();
						boolean doExtractDelete = this.config.isDoExtractDelete();
						Set<String> primaryKeyChecker = new HashSet<String>();
						if (doExtractDelete) {
							List<DDLPrimaryKey> primaryKeys = table.getPrimaryKeys();
							if (ValidateCommons.isNotEmpty(primaryKeys)) {
								primaryKeyChecker.addAll(primaryKeys.get(0).getPkColumnNames());
							}
						}

						for (DDLColumn column : columns) {
							String columnName = column.getColumnName();
							String dataType = column.getDataType();
							if (dialectCommons.TYPE.CLOB.equals(dataType) || dialectCommons.TYPE.NCLOB.equals(dataType) || 
									dialectCommons.TYPE.BLOB.equals(dataType)) {
								// 大字段
								lobColumns.add(column);
								continue;
							}
							if (insertColumnNames.length() > 0) {
								if (primaryKeyChecker.contains(columnName)) {
									primaryKeyWheres.append(" and ");
								}
								insertColumnNames.append(", ");
								selectColumnNames.append(", ");
								columnValues.append(", ");
							}
							if (dialectCommons.TYPE.CHAR.equals(dataType) || dialectCommons.TYPE.NCHAR.equals(dataType) || 
									dialectCommons.TYPE.VARCHAR.equals(dataType) || dialectCommons.TYPE.NVARCHAR.equals(dataType)) {
								// 字符串
								String columnValue = StringCommons.merger("'#[", columnName, "]'");
								if (primaryKeyChecker.contains(columnName)) {
									primaryKeyWheres.append(columnName).append(" = ").append(columnValue);
								}
								insertColumnNames.append(columnName);
								selectColumnNames.append(columnName);
								columnValues.append(columnValue);
							} else if (dialectCommons.TYPE.TIMESTAMP.equals(dataType)) {
								// 时间戳
								String columnValue = dialectCommons.FUNC.stringToDate(StringCommons.merger("'#[", columnName, "]'"), dialectCommons.FORMAT.TIMESTAMP23);
								if (primaryKeyChecker.contains(columnName)) {
									primaryKeyWheres.append(columnName).append(" = ").append(columnValue);
								}
								insertColumnNames.append(columnName);
								selectColumnNames.append(dialectCommons.FUNC.dateToString(columnName, dialectCommons.FORMAT.TIMESTAMP23)).append(" as ").append(columnName);
								columnValues.append(columnValue);
							} else if (dialectCommons.TYPE.DATETIME.equals(dataType)) {
								// 日期时间
								String columnValue = dialectCommons.FUNC.stringToDate(StringCommons.merger("'#[", columnName, "]'"), dialectCommons.FORMAT.DATETIME19);
								if (primaryKeyChecker.contains(columnName)) {
									primaryKeyWheres.append(columnName).append(" = ").append(columnValue);
								}
								insertColumnNames.append(columnName);
								selectColumnNames.append(dialectCommons.FUNC.dateToString(columnName, dialectCommons.FORMAT.DATETIME19)).append(" as ").append(columnName);
								columnValues.append(columnValue);
							} else if (dialectCommons.TYPE.DATE.equals(dataType)) {
								// 日期
								String columnValue = dialectCommons.FUNC.stringToDate(StringCommons.merger("'#[", columnName, "]'"), dialectCommons.FORMAT.DATE10);
								if (primaryKeyChecker.contains(columnName)) {
									primaryKeyWheres.append(columnName).append(" = ").append(columnValue);
								}
								insertColumnNames.append(columnName);
								selectColumnNames.append(dialectCommons.FUNC.dateToString(columnName, dialectCommons.FORMAT.DATE10)).append(" as ").append(columnName);
								columnValues.append(columnValue);
							} else {
								// 数字
								String columnValue = StringCommons.merger("#[", columnName, "]");
								if (primaryKeyChecker.contains(columnName)) {
									primaryKeyWheres.append(columnName).append(" = ").append(columnValue);
								}
								insertColumnNames.append(columnName);
								selectColumnNames.append(columnName);
								columnValues.append(columnValue);
							}
						}

						List<String> sqlTemplate = new ArrayList<String>();
						if (doExtractDelete && primaryKeyWheres.length() > 0) {
							sqlTemplate.add(StringCommons.merger("delete from ", toTableName, " where ", primaryKeyWheres));
						}
						sqlTemplate.add(StringCommons.merger("insert into ", toTableName, "(", insertColumnNames, ") values (", columnValues, ")"));
						SqlIOConfig dataIOConfig = new SqlIOConfig();
						dataIOConfig.setSqlTemplate(sqlTemplate);
						dataIOConfig.setOutputExpression(StringCommons.merger("select ", selectColumnNames, " from ", fromTableName, " ", this.config.getDataFliter(tableKey)));
						SqlIOTemplate dataIO = new SqlIOTemplate(JSON.toJSONString(dataIOConfig), dataIOConfig.getCharset());
						dataIO.output(dataIOSource, this.dataFolderPath, toTableName);
					}

					if (ValidateCommons.isNotEmpty(lobColumns)) {
						// 开始处理大字段
						StringBuffer columnNames = new StringBuffer();
						List<DDLPrimaryKey> primaryKeys = table.getPrimaryKeys();
						// 表必须存在主键才能对大字段进行 update
						if (ValidateCommons.isNotEmpty(primaryKeys)) {
							MemoryIOSource bigDataIndexIOSource = new MemoryIOSource();

							Set<String> primaryKeyChecker = new HashSet<String>(primaryKeys.get(0).getPkColumnNames());
							for (String primaryKey : primaryKeyChecker) {
								if (columnNames.length() > 0) {
									columnNames.append(", ");
								}
								columnNames.append(primaryKey);
							}
							for (DDLColumn column : lobColumns) {
								columnNames.append(", ").append(column.getColumnName());
							}
							String sql = StringCommons.merger("select ", columnNames, " from ", fromTableName);
							int pageIndex = 1;
							Page<Map<String, Object>> page = jdbcCommons.queryForPage(sql, null, pageIndex, BIG_DATA_PAGE_LIMIT);
							boolean isLast = false;
							do {
								isLast = page.isLast();
								for (Map<String, Object> result : page.getResults()) {
									Map<String, Object> params = new HashMap<String, Object>();
									for (String primaryKey : primaryKeyChecker) {
										params.put(primaryKey, result.get(primaryKey));
									}
									String updateWhere = PlaceholderCommons.replacePlaceholder(primaryKeyWheres.toString(), params);
									for (DDLColumn column : lobColumns) {
										String columnName = column.getColumnName();
										if (!primaryKeyChecker.contains(columnName)) {
											Object obj = result.get(columnName);
											if (ValidateCommons.isNotEmpty(obj)) {
												String updateStatement = StringCommons.merger("update ", toTableName, " set ", columnName, " = ? where ", updateWhere);
												String fileName = StringCommons.merger(UUID.randomUUID().toString(), ".", ENLARGE_NAME_BIG_DATA);
												String lobType = "";
												if (obj instanceof byte[]) {
													lobType = dialectCommons.TYPE.BLOB;
													StreamCommons.binaryTransport(new ByteArrayInputStream((byte[]) obj), 
															new FileOutputStream(FileCommons.appendPathNode(this.dataFolderPath, fileName)));
												} else if (obj instanceof String) {
													lobType = dialectCommons.TYPE.CLOB;
													StreamCommons.binaryTransport(new StringInputStream((String) obj), 
															new FileOutputStream(FileCommons.appendPathNode(this.dataFolderPath, fileName)));
												}
												bigDataIndexIOSource.setOutputResult(IO_EXPRESSION_BIG_DATA_INDEX, null, new ParamMap<String, Object>()
														.putParam(BIG_DATA_INDEX_KEY_UPDATE_STATEMENT, updateStatement)
														.putParam(BIG_DATA_INDEX_KEY_FILE_NAME, fileName)
														.putParam(BIG_DATA_INDEX_KEY_LOB_TYPE, lobType));
											}
										}
									}
								}
								pageIndex++;
								page = jdbcCommons.queryForPage(sql, null, pageIndex, BIG_DATA_PAGE_LIMIT);
							} while(!isLast);

							JSONIOTemplate bigDataIndexIO = new JSONIOTemplate(JarFileCommons.getResourceStream(BIG_DATA_INDEX_FILE_TEMPLATE));
							bigDataIndexIO.output(bigDataIndexIOSource, this.dataFolderPath, toTableName);
						}
					}
				}
			}
		}
	}

}
